/**
 * Created by zhangjinxia on 2017/10/12.
 */

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import jedis.Redis;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import util.PropertyUtil;
import java.util.ArrayList;
import java.util.List;

public class CanalClient {

    private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(CanalClient.class);
    private static Logger errorLogger = LoggerFactory.getLogger("canalClientErrorLogger");

    //我接的零工
    private static final String ORDER_USERID_STATUS099_LIST = "order-userid-status099-list";
    private static final String ORDER_USERID_STATUS020_LIST = "order-userid-status020-list";
    private static final String ORDER_USERID_STATUS030_LIST = "order-userid-status030-list";
    private static final String ORDER_USERID_STATUS040_LIST = "order-userid-status040-list";

    //我发的零工
    private static final String JOB_USERID_STATUS099_LIST = "job-userid-status099-list";
    private static final String JOB_USERID_STATUS020_LIST = "job-userid-status020-list";
    private static final String JOB_USERID_STATUS030_LIST = "job-userid-status030-list";
    private static final String JOB_USERID_STATUS040_LIST = "job-userid-status040-list";

    //任务的订单列表
    private static final String JOBORDER_JOBID_STATUS099_LIST = "joborder-jobid-status099-list";
    private static final String JOBORDER_JOBID_STATUS006_LIST = "joborder-jobid-status006-list";
    private static final String JOBORDER_JOBID_STATUS007_LIST = "joborder-jobid-status007-list";
    private static final String JOBORDER_JOBID_STATUS010_LIST = "joborder-jobid-status010-list";
    private static final String JOBORDER_JOBID_STATUS021_LIST = "joborder-jobid-status021-list";
    private static final String JOBORDER_JOBID_STATUS022_LIST = "joborder-jobid-status022-list";
    private static final String JOBORDER_JOBID_STATUS030_LIST = "joborder-jobid-status030-list";
    private static final String JOBORDER_JOBID_STATUS040_LIST = "joborder-jobid-status040-list";
    private static final String JOBORDER_JOBID_STATUS050_LIST = "joborder-jobid-status050-list";

    //订单信息
    private static final String ORDER_ORDERID_MAP = "order-orderid-map";
//    private static final String ORDER_ORDERID__EXPAND_MAP = "order-orderid-expand-map";

    //任务信息
    private static final String JOB_JOBID_MAP = "job-jobid-map";
//    private static final String JOB_JOBID_EXPAND_MAP = "job-jobid-expand-map";

    //用户信息
    private static final String USER_USERID_EXPAND_MAP = "user-userid-expand-map";
    private static final String USER_USERID_MAP = "user-userid-map";


    public static void main(String args[]) {
        // 创建链接
        CanalConnector connector = CanalConnectors.newClusterConnector(PropertyUtil.getValue("canal.zkServers"),PropertyUtil.getValue("canal.client.destination"),PropertyUtil.getValue("canal.username"),PropertyUtil.getValue("canal.password"));
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
//            connector.subscribe("dlg\\,dlg.oj_user,dlg.oj_order,dlg.oj_order_time,dlg.oj_order_evaluate");
            connector.subscribe("oddjob\\,oddjob.oj_user,oddjob.oj_order,oddjob.oj_order_time,oddjob.oj_job_task,oddjob.oj_job_task_time,oddjob.oj_user_personal_attribute,oddjob.oj_user_enterprise_attribute,oddjob.oj_order_evaluate");

            connector.rollback();
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    LOGGER.info("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    manageRedisEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } finally {
            connector.disconnect();
        }
    }

    private static void manageRedisEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }

            EventType eventType = rowChage.getEventType();
            LOGGER.info(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (RowData rowData : rowChage.getRowDatasList()) {
                try {
                    if (eventType == EventType.DELETE) {
                        String tableName = entry.getHeader().getTableName();
                        delete(rowData.getBeforeColumnsList(),tableName);
                        printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == EventType.INSERT) {
                        String tableName = entry.getHeader().getTableName();
                        insert(rowData, tableName);
                        printColumn(rowData.getAfterColumnsList());
                    } else if (eventType == EventType.UPDATE){
                        LOGGER.info("-------> before");
                        printColumn(rowData.getBeforeColumnsList());
                        LOGGER.info("-------> after");
                        printColumn(rowData.getAfterColumnsList());
                        String tableName = entry.getHeader().getTableName();
                        update(rowData.getBeforeColumnsList(),rowData.getAfterColumnsList(), tableName);
                    }
                }catch (Exception e){
                    errorLogger.error("ERROR",e);
                    errorLogger.error("-------> before");
                    printColumn(rowData.getBeforeColumnsList());
                    errorLogger.error("-------> after");
                    printColumn(rowData.getAfterColumnsList());
                    continue;
                }

            }
        }
    }

    private static void update(List<Column> beforeColumns, List<Column> afterColumns, String tableName) {
        if ("oj_order".equals(tableName)){//我接的零工
            updateOrder(beforeColumns, afterColumns);
        }else if("oj_order_time".equals(tableName)){//订单时间
            updateOrderTime(afterColumns);
        }else if("oj_job_task".equals(tableName)){//我发的零工
            updateJob(beforeColumns, afterColumns);
        }else if("oj_job_task_time".equals(tableName)){//任务时间
            updateJobTime(afterColumns);
        }else if("oj_user".equals(tableName)){//用户
            insertOrUpdateUser(afterColumns);
        }else if("oj_user_personal_attribute".equals(tableName)){//个人用户
            insertOrUpdateUserPersonal(afterColumns);
        }else if("oj_user_enterprise_attribute".equals(tableName)){//企业用户
            insertOrUpdateUserEnterprise(afterColumns);
        }else if("oj_order_evaluate".equals(tableName)){//订单评价
            insertOrUpdateOrderEvaluate(afterColumns);
        }
    }

    private static void updateJob(List<Column> beforeColumns, List<Column> afterColumns) {
        String active = "";//修改后的active
        String status = "";//修改后的状态
        String userId = "";//用户id
        String jobId = "";//任务id
        for (Column column:afterColumns) {
            if ("active".equals(column.getName()) && column.getUpdated()) {
                active = column.getValue();
            }else if("id".equals(column.getName())){
                jobId = column.getValue();
            }else if("status".equals(column.getName()) && column.getUpdated()){
                status = column.getValue();
            }else if("user_id".equals(column.getName())){
                userId = column.getValue();
            }
        }
        if (active.equals("0")){//任务改为无效
            deleteJob(afterColumns);
        }else if (active.equals("1")){//任务改为有效
            insertJob(afterColumns);
        }else {
            String beforeStatus = "";//修改前的状态
            if (StringUtils.isNotEmpty(status)) {//修改了任务状态
                for (Column column:beforeColumns) {
                    if("status".equals(column.getName())){
                        beforeStatus = column.getValue();
                        break;
                    }
                }
                //如果修改前和修改后的订单状态不一样
                if (!beforeStatus.equals(status)){
                    //我发的零工,状态有 进行中、已完成、待付款
                    if (beforeStatus.equals("20") || beforeStatus.equals("30") || beforeStatus.equals("40")) {
                        Redis.lrem(JOB_USERID_STATUS099_LIST.replace("99", beforeStatus).replace("userid", userId), 1, jobId);
                    }
                    if (status.equals("20") || status.equals("30") || status.equals("40")) {
                        Redis.lpush(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId), jobId);
                    }
                }
            }
            //最新更新的任务要放到最前面
            Redis.lrem(JOB_USERID_STATUS099_LIST.replace("userid", userId), 1, jobId);
            Redis.lpush(JOB_USERID_STATUS099_LIST.replace("userid", userId), jobId);
            if (status.equals("20") || status.equals("30") || status.equals("40")) {
                Redis.lrem(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId), 1, jobId);
                Redis.lpush(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId), jobId);
            }
            //更新任务信息
            for (Column column:afterColumns){
                Redis.hset(JOB_JOBID_MAP.replace("jobid",jobId),column.getName(), column.getValue());
            }
        }
    }

    private static void updateOrder(List<Column> beforeColumns, List<Column> afterColumns) {
        String active = "";//修改后的active
        String orderId = "";//订单id
        String status = "";//修改后的status
        String isInProgress = "";//修改后的是否是进行中
        String employeeId = "";//雇员id
        String jobId = "";//任务id
        for (Column column:afterColumns) {
            if ("active".equals(column.getName()) && column.getUpdated()) {
                active = column.getValue();
            }else if("id".equals(column.getName())){
                orderId = column.getValue();
            }else if("status".equals(column.getName()) && column.getUpdated()){
                status = column.getValue();
            }else if("is_in_progress".equals(column.getName())){
                isInProgress = column.getValue();
            }else if("employee_id".equals(column.getName())){
                employeeId = column.getValue();
            }else if("job_id".equals(column.getName())){
                jobId = column.getValue();
            }
        }
        String jobOrderVal = orderId+"-"+employeeId;
        String jobOrderAfterKeyStatus = getJobOrderStatus(status);

        if (active.equals("0")){
            deleteOrder(afterColumns);
        }else if (active.equals("1")){
            insertOrder(afterColumns);
        }else {
            String redisOrderAfterKeyStatus = "";//修改后的redis中key的status
            String redisOrderBeforeKeyStatus = "";//修改前的redis中key的status
            if (StringUtils.isNotEmpty(status)) {//修改了订单状态
                String beforeStatus = "";//修改前的状态
                String beforeIsInProgress = "";//修改前的是否是进行中
                for (Column column:beforeColumns) {
                    if("status".equals(column.getName())){
                        beforeStatus = column.getValue();
                    }else if("is_in_progress".equals(column.getName())){
                        beforeIsInProgress = column.getValue();
                    }
                }


                if (beforeIsInProgress.equals("1") && !beforeStatus.equals("30") && !beforeStatus.equals("40")){
                    redisOrderBeforeKeyStatus = "020";
                }else if (beforeStatus.equals("30")){
                    redisOrderBeforeKeyStatus = "030";
                }else if (beforeStatus.equals("40")){
                    redisOrderBeforeKeyStatus = "040";
                }else if (status.equals("50")){
                    redisOrderBeforeKeyStatus = "050";
                }

                if (isInProgress.equals("1") && !status.equals("30") && !status.equals("40")){
                    redisOrderAfterKeyStatus = "020";
                }else if (status.equals("30")){
                    redisOrderAfterKeyStatus = "030";
                }else if (status.equals("40")){
                    redisOrderAfterKeyStatus = "040";
                }else if (status.equals("50")){
                    redisOrderAfterKeyStatus = "050";
                }

                //joborder 的状态
                String jobOrderBeforeKeyStatus = getJobOrderStatus(beforeStatus);


                //如果修改前和修改后的订单状态不一样
                if (!redisOrderBeforeKeyStatus.equals(redisOrderAfterKeyStatus)){
                    //我接的零工没有"已取消"的分类
                    if (StringUtils.isNotEmpty(redisOrderBeforeKeyStatus) && !redisOrderBeforeKeyStatus.equals("50")) {
                        Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderBeforeKeyStatus).replace("userid", employeeId), 1, orderId);
                    }
                    //我接的零工没有"已取消"的分类
                    if (StringUtils.isNotEmpty(redisOrderAfterKeyStatus) && !redisOrderAfterKeyStatus.equals("50")) {
                        Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("userid", employeeId), orderId);
                    }
                }
                if (!jobOrderBeforeKeyStatus.equals(jobOrderAfterKeyStatus)){
                    Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderBeforeKeyStatus).replace("jobid", jobId), 1, jobOrderVal);
                    Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderAfterKeyStatus).replace("jobid", jobId), jobOrderVal);

                }
            }

            //将最新更新的放在最前面
            Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("userid", employeeId), 1, orderId);
            Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("userid", employeeId), orderId);
            Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", jobId), 1, jobOrderVal);
            Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", jobId), jobOrderVal);

            //order的状态如果不为空，则将最新更新的放在最前面
            if (StringUtils.isNotEmpty(redisOrderAfterKeyStatus)){
                Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("userid", employeeId), 1, orderId);
                Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("userid", employeeId), orderId);
            }
            //joborder的状态如果不为空，则将最新更新的放在最前面
            if (StringUtils.isNotEmpty(jobOrderAfterKeyStatus)){
                Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("jobid", jobId), 1, jobOrderVal);
                Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("jobid", jobId), jobOrderVal);
            }

            //更新订单详情
            for (Column column : afterColumns) {
                if (column.getUpdated()) {
                    Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), column.getName(), column.getValue());
                }
            }
        }
    }

    private static String getJobOrderStatus(String status){
//        30,22,10,21(20),50,40,6(8),7(9)
        String result = "";
        if ("6".equals(status) || "8".equals(status)){
            result = "006";
        }else if("7".equals(status) || "9".equals(status)){
            result = "007";
        }else if("21".equals(status) || "20".equals(status)){
            result = "021";
        }else if("10".equals(status)){
            result = "010";
        }else if("30".equals(status)){
            result = "030";
        }else if("22".equals(status)){
            result = "022";
        }else if("50".equals(status)){
            result = "050";
        }else if("40".equals(status)){
            result = "040";
        }
        return result;
    }

    private static void delete(List<Column> columns, String tableName) {
        if ("oj_order".equals(tableName)){//删除订单
            deleteOrder(columns);
        }else if("oj_job_task".equals(tableName)){//删除任务
            deleteJob(columns);
        }else if("oj_user".equals(tableName)){//删除用户
            deleteUser(columns);
        }else if("oj_order_evaluate".equals(tableName)){//订单评价
            deleteOrderEvaluate(columns);
        }
    }

    private static void deleteOrderEvaluate(List<Column> columns) {
        String orderId = "";
        for (Column column:columns){
            if ("order_id".equals(column.getName())){
                orderId = column.getValue();
                break;
            }
        }
        if (StringUtils.isNotEmpty(orderId)) {
            Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), "is_evaluate", "0");
        }
    }

    private static void deleteUser(List<Column> columns) {
        String userId = "";
        for (Column column:columns){
            if (StringUtils.isNotEmpty(userId))
                break;
            if ("id".equals(column.getName())){
                userId = column.getValue();
            }
        }
        Redis.del(USER_USERID_EXPAND_MAP.replace("userid",userId));//用户扩展信息
        Redis.del(USER_USERID_MAP.replace("userid",userId));//用户信息
    }

    private static void deleteJob(List<Column> columns) {
        String status = "";
        String userId = "";
        String jobId = "";
        for (Column column : columns) {
            if ("status".equals(column.getName())){
                status = column.getValue();
            }else if ("user_id".equals(column.getName())){
                userId = column.getValue();
            }else if ("id".equals(column.getName())){
                jobId = column.getValue();
            }
        }
        String replaceStatus = "status0"+status;
        Redis.lrem(JOB_USERID_STATUS099_LIST.replace("userid",userId),1,jobId);//用户发布的任务
        //用户发布的任务
        Redis.lrem(JOB_USERID_STATUS099_LIST.replace("status099",replaceStatus).replace("userid",userId),1,jobId);
        Redis.del(JOBORDER_JOBID_STATUS099_LIST);//任务对应的订单列表
        Redis.del(JOBORDER_JOBID_STATUS006_LIST);//任务对应的订单列表
        Redis.del(JOBORDER_JOBID_STATUS030_LIST);//任务对应的订单列表
        Redis.del(JOBORDER_JOBID_STATUS040_LIST);//任务对应的订单列表
        Redis.del(JOBORDER_JOBID_STATUS050_LIST);//任务对应的订单列表
        Redis.del(JOBORDER_JOBID_STATUS007_LIST);//任务对应的订单列表
        Redis.del(JOBORDER_JOBID_STATUS010_LIST);//任务对应的订单列表
        Redis.del(JOBORDER_JOBID_STATUS021_LIST);//任务对应的订单列表
        Redis.del(JOBORDER_JOBID_STATUS022_LIST);//任务对应的订单列表

        Redis.del(JOB_JOBID_MAP.replace("jobid",jobId));//任务信息
//        Redis.del(JOB_JOBID_EXPAND_MAP.replace("jobid",jobId));//任务扩展信息
    }

    private static void deleteOrder(List<Column> columns) {
        String orderId = "";
        String jobId = "";
        String employeeId = "";
        for (Column column : columns) {
            if ("id".equals(column.getName())){
                orderId = column.getValue();
            }else if("employee_id".equals(column.getName())){
                employeeId = column.getValue();
            }else if("job_id".equals(column.getName())){
                jobId = column.getValue();
            }
        }
        String jobOrderVal = orderId+"-"+employeeId;

//        Redis.del(ORDER_ORDERID__EXPAND_MAP.replace("orderid",orderId));//订单扩展信息
        Redis.del(ORDER_ORDERID_MAP.replace("orderid",orderId));//订单信息
        Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("userid",employeeId),1, orderId);//用户接单列表
        Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("jobid",jobId),1,jobOrderVal);//任务对应的订单列表
        Redis.lrem(JOBORDER_JOBID_STATUS006_LIST.replace("jobid",jobId),1,jobOrderVal);//任务对应的订单列表
        Redis.lrem(JOBORDER_JOBID_STATUS030_LIST.replace("jobid",jobId),1,jobOrderVal);//任务对应的订单列表
        Redis.lrem(JOBORDER_JOBID_STATUS040_LIST.replace("jobid",jobId),1,jobOrderVal);//任务对应的订单列表
        Redis.lrem(JOBORDER_JOBID_STATUS050_LIST.replace("jobid",jobId),1,jobOrderVal);//任务对应的订单列表
        Redis.lrem(JOBORDER_JOBID_STATUS007_LIST.replace("jobid",jobId),1,jobOrderVal);//任务对应的订单列表
        Redis.lrem(JOBORDER_JOBID_STATUS021_LIST.replace("jobid",jobId),1,jobOrderVal);//任务对应的订单列表
        Redis.lrem(JOBORDER_JOBID_STATUS022_LIST.replace("jobid",jobId),1,jobOrderVal);//任务对应的订单列表
        Redis.lrem(JOBORDER_JOBID_STATUS010_LIST.replace("jobid",jobId),1,jobOrderVal);//任务对应的订单列表
    }

    private static void insert(RowData rowData, String tableName){
        List<Column> columns = rowData.getAfterColumnsList();
        if ("oj_order".equals(tableName)){//我接的零工
            insertOrder(columns);
        }else if("oj_order_time".equals(tableName)){//订单时间
            insertOrderTime(columns);
        }else if("oj_job_task".equals(tableName)){//我发的零工
            insertJob(columns);
        }else if("oj_job_task_time".equals(tableName)){//任务时间
            insertJobTime(columns);
        }else if("oj_user".equals(tableName)){//用户
            insertOrUpdateUser(columns);
        }else if("oj_user_personal_attribute".equals(tableName)){//个人用户
            insertOrUpdateUserPersonal(columns);
        }else if("oj_user_enterprise_attribute".equals(tableName)){//企业用户
            insertOrUpdateUserEnterprise(columns);
        }else if("oj_order_evaluate".equals(tableName)){//订单评价
            insertOrUpdateOrderEvaluate(rowData.getAfterColumnsList());
        }
    }

    private static void insertOrUpdateOrderEvaluate(List<Column> columns) {
        String orderId = "";
        String active = "";
        String accepter_user_id = "";
        String appraiser_user_id = "";
        String evaluate_level = "";
        for (Column column:columns){
            if ("order_id".equals(column.getName())){
                orderId = column.getValue();
            }else if("active".equals(column.getName())){//修改了active字段
                active = column.getValue();
            }else if("accepter_user_id".equals(column.getName())){//被评价人用户id
                accepter_user_id = column.getValue();
            }else if("appraiser_user_id".equals(column.getName())){//评价人用户id
                appraiser_user_id = column.getValue();
            }else if("evaluate_level".equals(column.getName())){
                evaluate_level = column.getValue();
            }
        }
        String employeeId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid", orderId), "employee_id");
        String employerId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid", orderId), "employer_id");

        if (StringUtils.isNotEmpty(orderId)) {
            if ("0".equals(active)) {//删除评价
                if (employeeId.equals(accepter_user_id)) {//被评价人是雇员
                    Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), "is_ee_evaluate", "0");
                    Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), "is_ee_evaluate_level", "0");
                }else if (employerId.equals(accepter_user_id)){//被评价人是雇主
                    Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), "is_er_evaluate", "0");
                    Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), "is_er_evaluate_level", "0");
                }
            } else if ("1".equals(active)) {//将评价改为有效
                if (employeeId.equals(accepter_user_id)){//被评价人是雇员
                    Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), "is_ee_evaluate", "1");
                    Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), "is_ee_evaluate_level", evaluate_level);
                }else if (employerId.equals(accepter_user_id)){//被评价人是雇主
                    Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), "is_er_evaluate", "1");
                    Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), "is_er_evaluate_level", evaluate_level);
                }
            }
        }
    }

    private static void insertOrUpdateUserEnterprise(List<Column> columns) {
        String userId = "";
        List<String> exUserMap = new ArrayList<>();
        exUserMap.add("logo");
        exUserMap.add("name");
        exUserMap.add("credit_count");
        exUserMap.add("score_count");
        exUserMap.add("short_name");
        for (Column column:columns){
            if ("user_id".equals(column.getName())){
                userId = column.getValue();
                break;
            }
        }
        for (Column column:columns){
            if (exUserMap.contains(column.getName()) && column.getUpdated()){
                Redis.hset(USER_USERID_EXPAND_MAP.replace("userid",userId),column.getName(), column.getValue());
            }
        }
    }

    private static void insertOrUpdateUserPersonal(List<Column> columns) {
        //头像、手机号码、用户名称、诚信值、评分值
        String userId = "";
        List<String> exUserMap = new ArrayList<>();
        exUserMap.add("logo");
        exUserMap.add("phone");
        exUserMap.add("name");
        exUserMap.add("credit_count");
        exUserMap.add("score_count");
        for (Column column:columns){
            if ("user_id".equals(column.getName())){
                userId = column.getValue();
                break;
            }
        }
        for (Column column:columns){
            if (exUserMap.contains(column.getName())){
                Redis.hset(USER_USERID_EXPAND_MAP.replace("userid",userId),column.getName(), column.getValue());
            }
        }
    }

    private static void insertOrUpdateUser(List<Column> columns) {
        String userId = "";
        for (Column column:columns) {
            if ("id".equals(column.getName())) {
                userId = column.getValue();
                break;
            }
        }
        for (Column column:columns){
            if ("username".equals(column.getName())){
                Redis.hset(USER_USERID_EXPAND_MAP.replace("userid",userId),column.getName(), column.getValue());
            }else if ("phone".equals(column.getName())){
                Redis.hset(USER_USERID_EXPAND_MAP.replace("userid",userId),column.getName(), column.getValue());
            }
        }
    }

    private static void insertOrder(List<Column> columns) {
        String orderId = "";
        String employeeId = "";
        String jobId = "";
        String status = "";
        for (Column column : columns) {
            if ("id".equals(column.getName())) {
                orderId = column.getValue();
            } else if ("job_id".equals(column.getName())) {
                jobId = column.getValue();
            } else if ("employee_id".equals(column.getName())) {
                employeeId = column.getValue();
            } else if ("status".equals(column.getName())) {
                status = column.getValue();
            }
        }
        for (Column column : columns) {
            Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), column.getName(), column.getValue());
        }
        Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("userid", employeeId), orderId);
        Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", jobId), orderId + "-" + employeeId);
        String jobOrderKeyStatus = getJobOrderStatus(status);
        if (StringUtils.isNotEmpty(jobOrderKeyStatus)) {
            Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderKeyStatus).replace("jobid", jobId), orderId + "-" + employeeId);
        }
    }

    private static void insertJob(List<Column> columns){
        //任务的扩展信息字段
        List<String> exJobMap = new ArrayList<>();
        exJobMap.add("active");
        exJobMap.add("is_farmers_insurance");
        exJobMap.add("status");

        String userId = "";
        String jobId = "";
        for (Column column:columns){
            if (StringUtils.isNotEmpty(userId) && StringUtils.isNotEmpty(jobId))
                break;
            if ("id".equals(column.getName())){
                jobId = column.getValue();
            }
            if ("user_id".equals(column.getName())){
                userId = column.getValue();
            }
        }
        for (Column column:columns){
            Redis.hset(JOB_JOBID_MAP.replace("jobid",jobId),column.getName(), column.getValue());
        }
        Redis.lpush(JOB_USERID_STATUS099_LIST.replace("userid",userId),jobId);
    }

    private static void updateOrderTime(List<Column> columns){
        String orderId = "";

        List<String> timeColumnName = getTimeColumns();
        for (Column column:columns) {
            if ("order_id".equals(column.getName())){
                orderId = column.getValue();
            }
        }
        for (Column column:columns) {
            if (timeColumnName.contains(column.getName())) {
                Redis.hset(ORDER_ORDERID_MAP.replace("orderid",orderId), column.getName(), column.getValue());
            }
        }

        String redisOrderKeyStatus = "";
        String isInProgress = Redis.hget(ORDER_ORDERID_MAP.replace("orderid",orderId),"is_in_progress");
        String status = Redis.hget(ORDER_ORDERID_MAP.replace("orderid",orderId),"status");
        String employeeId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid",orderId),"employee_id");
        String jobId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid",orderId),"job_id");
        String jobOrderVal = orderId+"-"+employeeId;

        if (isInProgress.equals("1") && !status.equals("30") && !status.equals("40")){
            redisOrderKeyStatus = "020";
        }else if (status.equals("30")){
            redisOrderKeyStatus = "030";
        }else if (status.equals("40")){
            redisOrderKeyStatus = "040";
        }else if (status.equals("50")){
            redisOrderKeyStatus = "050";
        }

        String jobOrderKeyStatus = getJobOrderStatus(status);

        Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("userid",employeeId),1, orderId);
        Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("userid",employeeId), orderId);
        Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("jobid",jobId),1, jobOrderVal);
        Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("jobid",jobId), jobOrderVal);
        if (StringUtils.isNotEmpty(redisOrderKeyStatus)) {
            if (!redisOrderKeyStatus.equals("050")) {
                Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderKeyStatus).replace("userid", employeeId), 1, orderId);
                Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderKeyStatus).replace("userid", employeeId), orderId);
            }
        }
        //最新更新的放前面
        if (StringUtils.isNotEmpty(jobOrderKeyStatus)){
            Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderKeyStatus).replace("jobid",jobId),1, jobOrderVal);
            Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderKeyStatus).replace("jobid",jobId), jobOrderVal);
        }
    }

    private static List<String> getTimeColumns(){
        List<String> timeColumnName = new ArrayList<>();
        timeColumnName.add("is_cross_day");
        timeColumnName.add("start_year");
        timeColumnName.add("start_month");
        timeColumnName.add("start_day");
        timeColumnName.add("start_hour");
        timeColumnName.add("start_minute");
        timeColumnName.add("start_second");
        timeColumnName.add("end_year");
        timeColumnName.add("end_month");
        timeColumnName.add("end_day");
        timeColumnName.add("end_hour");
        timeColumnName.add("end_minute");
        timeColumnName.add("end_second");
        return timeColumnName;
    }

    private static void insertOrderTime(List<Column> columns){
        String orderId = "";
        List<String> timeColumnName = getTimeColumns();
        for (Column column:columns) {
            if ("order_id".equals(column.getName())) {
                orderId = column.getValue();
            }
        }
        for (Column column:columns) {
            if (timeColumnName.contains(column.getName())) {
                Redis.hset(ORDER_ORDERID_MAP.replace("orderid",orderId), column.getName(), column.getValue());
            }
        }
    }

    private static void insertJobTime(List<Column> columns){
        String jobId = "";
        List<String> timeColumnName = getTimeColumns();
        //找到任务id
        for (Column column:columns) {
            if ("job_task_id".equals(column.getName())) {
                jobId = column.getValue();
            }
        }
        for (Column column:columns){
            if (timeColumnName.contains(column.getName())) {
                Redis.hset(JOB_JOBID_MAP.replace("jobid",jobId),column.getName(), column.getValue());
            }
        }
    }

    private static void updateJobTime(List<Column> columns){
        String jobId = "";
        List<String> timeColumnName = getTimeColumns();
        //找到任务id
        for (Column column:columns) {
            if ("job_task_id".equals(column.getName())) {
                jobId = column.getValue();
            }
        }
        for (Column column:columns){
            if (timeColumnName.contains(column.getName())) {
                Redis.hset(JOB_JOBID_MAP.replace("jobid",jobId),column.getName(), column.getValue());
            }
        }

        String status = Redis.hget(JOB_JOBID_MAP.replace("jobid",jobId),"status");
        String userId = Redis.hget(JOB_JOBID_MAP.replace("jobid",jobId),"user_id");
        Redis.lrem(JOB_USERID_STATUS099_LIST.replace("userid", userId),1,jobId);
        Redis.lpush(JOB_USERID_STATUS099_LIST.replace("userid", userId), jobId);
        if (status.equals("20") || status.equals("30") || status.equals("40")) {
            Redis.lrem(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId),1,jobId);
            Redis.lpush(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId), jobId);
        }
    }

    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            LOGGER.info(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }

    private static String getStart(List<Column> columns){
        String startYear="",startMonth="",startDay="",startHour="",startMinutes="",startSecond="";
        for (Column column:columns) {
            if ("start_year".equals(column.getName())) {
                startYear = column.getValue();
            } else if ("start_month".equals(column.getName())) {
                startMonth = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            } else if ("start_day".equals(column.getName())) {
                startDay = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            } else if ("start_hour".equals(column.getName())) {
                startHour = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            } else if ("start_minute".equals(column.getName())) {
                startMinutes = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            } else if ("start_second".equals(column.getName())) {
                startSecond = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            }
        }
        return new StringBuilder(startYear).append("-").append(startMonth).append("-").append(startDay).append(" ").append(startHour).append(":").append(startMinutes).append(":").append(startSecond).toString();
    }

    private static String getEnd(List<Column> columns){
        String endYear="",endMonth="",endDay="",endHour="",endMinutes="",endSecond="";
        for (Column column:columns) {
            if ("end_year".equals(column.getName())){
                endYear = column.getValue();
            }else if ("end_month".equals(column.getName())){
                endMonth = StringUtils.isEmpty(column.getValue()) ? null : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }else if ("end_day".equals(column.getName())){
                endDay = StringUtils.isEmpty(column.getValue()) ? null : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }else if ("end_hour".equals(column.getName())){
                endHour = StringUtils.isEmpty(column.getValue()) ? "00" : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }else if ("end_minute".equals(column.getName())){
                endMinutes = StringUtils.isEmpty(column.getValue()) ? "00" : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }else if ("end_second".equals(column.getName())){
                endSecond = StringUtils.isEmpty(column.getValue()) ? "00" : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }
        }
        if (endYear!=null && endMonth!=null && endDay!=null) {
            return new StringBuilder(endYear).append("-").append(endMonth).append("-").append(endDay).append(" ").append(endHour).append(":").append(endMinutes).append(":").append(endSecond).toString();
        }else{
            return "";
        }
    }
}